package cn.hy.dataSync.job.scheduler;




import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.statistics.container.communicator.AbstractContainerCommunicator;
import cn.hy.dataSync.statistics.container.communicator.TaskContainerCommunicator;
import cn.hy.dataSync.task.TaskContainer;
import cn.hy.dataSync.task.runner.TaskContainerRunner;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 父类的部分实现
 */
public abstract class ProcessScheduler extends AbstractScheduler {

    private ExecutorService taskContainerExecutorService;

    private AbstractContainerCommunicator containerCommunicator;

    public ProcessScheduler(AbstractContainerCommunicator containerCommunicator) {
        super(containerCommunicator);
        this.containerCommunicator = containerCommunicator;
    }

    /**
     * 提交所有任务到线程池
     * @param taskConfigs 任务配置
     */
    @Override
    public void startAllTask(final List<TaskConfig> taskConfigs) {
        this.taskContainerExecutorService = Executors.newFixedThreadPool(taskConfigs.size());

        for (TaskConfig taskConfig : taskConfigs) {
            // 每个任务用一个线程执行
            TaskContainerRunner taskContainerRunner = newTaskContainerRunner(taskConfig);
            this.taskContainerExecutorService.execute(taskContainerRunner);
        }

        // 直接关闭，会等待线程池中所有任务执行完毕后关闭
        this.taskContainerExecutorService.shutdown();
    }

    /**
     * 如果失败，直接关闭线程池 剩余任务被中断 丢弃
     * @param frameworkCollector 状态搜集器
     * @param throwable          异常信息
     */
    @Override
    public void dealFailedStat(AbstractContainerCommunicator frameworkCollector, Throwable throwable) {
        this.taskContainerExecutorService.shutdownNow();
        throw new RuntimeException(throwable);
    }

    /**
     *
     * @param frameworkCollector 状态搜集器
     * @param totalTasks         任务数
     */
    @Override
    public void dealKillingStat(AbstractContainerCommunicator frameworkCollector, int totalTasks) {
        this.taskContainerExecutorService.shutdownNow();
        throw new RuntimeException("job killed status");
    }

    /**
     *
     * @param taskConfig 任务配置
     * @return
     */
    private TaskContainerRunner newTaskContainerRunner(TaskConfig taskConfig) {
        //containerCommunicator.getReporter() 公共一个
        TaskContainerCommunicator taskContainerCommunicator =
                new TaskContainerCommunicator(taskConfig, containerCommunicator.getReporter());
        TaskContainer taskContainer = new TaskContainer(taskConfig, taskContainerCommunicator);
        return new TaskContainerRunner(taskContainer);
    }
}
